package com.zhuwei.kafka.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.List;

/**
 * 功能：kafka消息发送，kafkaTemplate实现
 * @author zhuiwei
 * @version 2019/05/14
 */
@Component
public class KafKaProducer {
    private final static Logger log= LoggerFactory.getLogger(KafKaProducer.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;
    /**
     * 发送消息
     * @param topic 主题
     * @param object 数据对象
     */
    public void sendMessage(String topic,Object object) {
            ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(topic, object );
            //回调处理
            send.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    log.info("发生消息失败：" + throwable.getMessage());
                }

                @Override
                public void onSuccess(@Nullable SendResult<String, Object> stringObjectSendResult) {
                    log.info("发生消息成功：" + stringObjectSendResult.toString());
                }
            });
    }

    /**
     * 随机将数组中的数据发放到各个分区
     * @param topic 主题
     * @param list 消息数组
     * @param partitionNum 分区数
     */
    public void sendMessage(String topic, List<String> list, Integer partitionNum) {
        for (int i = 0; i < list.size(); i++) {
            ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(topic, i % partitionNum, "key", list.get(i)+i );
            //回调处理
            send.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    log.info("发生消息失败：" + throwable.getMessage());
                }

                @Override
                public void onSuccess(@Nullable SendResult<String, Object> stringObjectSendResult) {
                    log.info("发生消息成功：" + stringObjectSendResult.toString());
                }
            });
        }
    }
}
